package com.example;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

@Slf4j
public class ProducerImpl implements Producer {
    private final KafkaProducer<String, String> producer;
    private final String topic = "test32";

    public ProducerImpl() {
        producer = new KafkaProducer<>(props());
    }

    private Properties props() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9001");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }

    @Override
    public void send(Order order) {
        log.info("{}", order);
        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, order.getId().toString(), JSON.toJSONString(order));
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                } else {
                    log.info("callback -> metadata={}", metadata);
                }
            }).get();
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }

    @Override
    public void waitComplete() {
        producer.flush();
    }
}
